package com.atguigu.flink.chapter10;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2023/6/27 09:23
 */
public class TableApiDemo1 {
    public static void main(String[] args) throws Exception {
        // 1. 表的执行
        /*EnvironmentSettings conf = EnvironmentSettings.newInstance()
            .inStreamingMode()
            .build();
        TableEnvironment tEnv = TableEnvironment.create(conf);*/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s2", 2000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40),
            new WaterSensor("s2", 5000L, 50)
        );
    
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
        Table table = tEnv.fromDataStream(stream);
        table.printSchema();
        
        // api 的方式: select * from t where id='s1'
        Table result = table
            .where($("id").isEqual("s1"))
            .select($("id"), $("ts"), $("vc"));
    
        
        // 把表转成流
        DataStream<WaterSensor> resultStream = tEnv.toDataStream(result, WaterSensor.class);
        resultStream.print();
    
        env.execute();
    }
}
